package com.foreveross.bsl.common.utils.balance.zookeeper2;

import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.foreveross.bsl.common.utils.balance.ServiceKeeper;
import com.foreveross.bsl.common.utils.balance.ServiceListener;

/**
 * ZK定制启动程序
 *
 * @author hyw
 */
public class ZKServiceKeeper extends ZKCommonKeeper implements ServiceKeeper {

    private static final long serialVersionUID = 1256501851252192726L;

    private static final Logger logger = LoggerFactory.getLogger(ZKServiceKeeper.class);
    
    ServiceListener listener; // 单节点监听器
    Map<String, ServiceListener> pathListenerMap = new HashMap<String, ServiceListener>(); // 多节点监听器

    
    public ZKServiceKeeper(String serverList, int sessionTimeout) {
        super(serverList, sessionTimeout);
    }
    
    @Override
    public void register(String path, String data, CreateMode createMode) {
        this.path = path;
        try {
        	ZKUtil.buildPath(zk, path, (createMode == null ? CreateMode.PERSISTENT : createMode)); // 最后一级节点选择持久方式
            Stat stat = zk.exists(path, false);
            if(stat == null) { // 不存在节点
                zk.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); // 此节点临时
            }
            zk.setData(path, data.getBytes("UTF-8"), stat.getVersion());
        } catch (KeeperException e) {
        	logger.error("zookeeper连接失效！", e);
        } catch (InterruptedException e) {
        	logger.error("zookeeper中断异常！", e);
        } catch (UnsupportedEncodingException e) {
        	logger.error("此节点数据有问题!" + path , e);
        }
    }

    @Override
    public void register(String serviceParentPath, String ip, String data, CreateMode createMode) {
        if (ip.contains("/")) {
            throw new IllegalArgumentException("ip参数不能包含/");
        }
        this.path = serviceParentPath;
        try {
        	 ZKUtil.buildPath(zk, serviceParentPath, (createMode == null ? CreateMode.PERSISTENT : createMode));
             Stat stat = zk.exists(serviceParentPath, false);
            if (stat != null) {
                String servicePath = serviceParentPath + "/" + ip;
                zk.create(serviceParentPath + "/" + ip, null, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                zk.setData(servicePath, data.getBytes("UTF-8"), stat.getVersion());
            } else {
                throw new RuntimeException("没有合法的aliveServicePath");
            }
        } catch (KeeperException e) {
        	logger.error("zookeeper连接失效！", e);
        } catch (InterruptedException e) {
        	logger.error("zookeeper中断异常！", e);
        } catch (UnsupportedEncodingException e) {
        	logger.error("此节点数据有问题!" + path , e);
        }
    }

    @Override
    public String getData(String path) {
    	String result = null;
        try {
			return ZKUtil.getData(zk, path);
		} catch (Exception e) {
			logger.error("zookeeper异常！", e);
		} 
        return result;
    }
    
    @Override
    public String getData(String serviceParentPath, String ip) {
    	String result = null;
        try {
			return ZKUtil.getData(zk, serviceParentPath + "/" + ip);
		} catch (Exception e) {
			logger.error("zookeeper异常！", e);
		}
        return result;
    }

    @Override
    public Map<String, String> getServiceMap(String serviceParentPath) {
        Map<String, String> serviceMap = new HashMap<String, String>();
        try {
            if (zk.exists(serviceParentPath, false) != null) {
                List<String> serviceList = zk.getChildren(serviceParentPath, false);
                for (String ip : serviceList) {
                    serviceMap.put(ip, ZKUtil.getData(zk, serviceParentPath + "/" + ip));
                }
            }
        } catch (KeeperException e) {
        	logger.error("zookeeper连接失效！", e);
        } catch (InterruptedException e) {
        	logger.error("zookeeper中断异常！", e);
        } catch (Exception e) {
        	logger.error("此节点数据有问题!" + path , e);
        }
        return serviceMap;
    }
    
    @Override
    public void listen(Map<String, ServiceListener> listenerMap) {
    	 if (pathListenerMap == null) {
			throw new IllegalArgumentException("pathListenerMap不能为空！");
		}
		this.pathListenerMap = listenerMap;
		this.paths = pathListenerMap.keySet();
		bindWatcher(this.paths);
    }

    @Override
    public void listen(String serviceParentPath, ServiceListener listener) {
        this.path = serviceParentPath;
        this.listener = listener;
        this.bindWatcher(serviceParentPath);
    }

    void bindWatcher(String serviceParentPath) {
    	if(StringUtils.isBlank(serviceParentPath))
    		return;
        try {
            ZKUtil.buildPath(zk, serviceParentPath, CreateMode.PERSISTENT); // 存在节点不会创建
            // 为某个节点添加watch
            zk.getChildren(serviceParentPath, new ZKCommonWatcher(this)); 
        } catch (KeeperException e) {
        	logger.error("zookeeper连接失效！", e);
        } catch (InterruptedException e) {
        	logger.error("zookeeper中断异常！", e);
        }
    }

    void bindWatcher(Set<String> servicePaths) {
        for (String servicePath : servicePaths) {
            try {
                ZKUtil.buildPath(zk, servicePath, CreateMode.PERSISTENT);
                zk.getChildren(servicePath, new ZKCommonWatcher(this, servicePath));
            } catch (KeeperException e) {
            	logger.error("zookeeper连接失效！", e);
            } catch (InterruptedException e) {
            	logger.error("zookeeper中断异常！", e);
            }
        }
    }

    void process() {
		for (String path : this.pathListenerMap.keySet()) {
			ServiceListener listener = pathListenerMap.get(path);
			listener.process(this.getServiceMap(path));
		}
        if (this.listener != null) {
            this.listener.process(this.getServiceMap(path));
        } else {
           logger.warn("第一次连接，监听器为空！");
        }
    }

    @Override
    void afterConnected() {
        logger.info("重新获取获取连接。。。");
        for (String path : this.pathListenerMap.keySet()) {
			ServiceListener listener = pathListenerMap.get(path);
			listener.afterConnected();
		}
        if (this.listener != null) {
            this.listener.afterConnected();
        } else {
           logger.warn("第一次连接，监听器为空！");
        }
        process();
    }

}
